package org.gxlu.com.storm.wc;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

@SuppressWarnings("serial")
public class SplitBolt  extends BaseRichBolt {

	private OutputCollector collector = null ;
	
	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector_) {
		collector  = collector_ ;
		
	}

	@Override
	public void execute(Tuple input) {
		 String line = input.getString(0) ;
		 String[] array = line.split(" ") ;
		 for(String a:array){
			 if(a==null || a.length() == 0){
				 continue ;
			 }
			 collector.emit(new Values(a,1));
		 }
		// collector.ack(input);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word","num")); 
	}
 

}
